package com.example.demo.controller;

import com.example.demo.bean.KafkaLog;
import com.example.demo.constant.KafkaConstant;
import com.example.demo.domain.Message;
import com.example.demo.mapper.KafkaLogMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;

/**
 * @author MrBird
 */
@RestController
public class SendMessageController {

    private Logger logger = LoggerFactory.getLogger(this.getClass());
    @Autowired
    // private KafkaTemplate<String, String> kafkaTemplate;
    private KafkaTemplate<String, Message> kafkaTemplate;
    @Autowired
    KafkaLogMapper kafkaLogMapper;

    @GetMapping("send/{message}/{messageId}")
    public void send(@PathVariable("message") String message, @PathVariable("messageId") String messageId) {
        KafkaLog kafkaLog = getKafkaLog(message, messageId);
        addKafkaLog(kafkaLog);
        ListenableFuture<SendResult<String, Message>> future = this.kafkaTemplate.send(KafkaConstant.topic, new Message(kafkaLog.getMessageId(), message));
        future.addCallback(new ListenableFutureCallback<SendResult<String, Message>>() {
            @Override
            public void onSuccess(SendResult<String, Message> result) {
                logger.info("成功发送消息：{}，offset=[{}]", message, result.getRecordMetadata().offset());
                kafkaLogMapper.update(kafkaLog.getMessageId(), KafkaConstant.sendSuccess);
            }

            @Override
            public void onFailure(Throwable ex) {
                logger.error("消息：{} 发送失败，原因：{}", message, ex.getMessage());
                kafkaLogMapper.update(kafkaLog.getMessageId(), KafkaConstant.sendFail);
            }
        });
    }

//    @GetMapping("send/{message}")
//    public void sendMessage(@PathVariable String message) {
//        this.kafkaTemplate.send("test", new Message("mrbird", message));
//    }

    private KafkaLog getKafkaLog(String message, String messageId) {
        KafkaLog kafkaLog = new KafkaLog();
        kafkaLog.setMessageId(messageId);
        kafkaLog.setMessageBody(message);
        kafkaLog.setTopic(KafkaConstant.topic);
        kafkaLog.setStatus(KafkaConstant.sendIng);
        kafkaLog.setNum("0");
        kafkaLog.setSend("0");
        return kafkaLog;
    }

    private void addKafkaLog(KafkaLog kafkaLog) {
        KafkaLog queryKafkaLog = kafkaLogMapper.queryKafkaLogByMessageId(kafkaLog.getMessageId());
        if (queryKafkaLog == null) {
            kafkaLogMapper.add(kafkaLog);
        } else {
            kafkaLogMapper.updateNumAndStatus(kafkaLog.getMessageId(), KafkaConstant.sendIng);
        }
    }
}
